package com.atguigu.datastream.test.day04;

import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

/**
 * ClassName: Flink_01_Sink_File
 * Package: com.atguigu.test.day04
 * Description:
 *
 * @Author ChenJun
 * @Create 2023/4/10 20:59
 * @Version 1.0
 */
public class Flink_01_Sink_File {
    public static void main(String[] args) throws Exception {

        //1. 创建流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2. 读取文件
        DataStreamSource<String> streamSource = env.fromElements("chenjun nihao,kangshifu yyds");

        //创建sink
        StreamingFileSink<String> sink = StreamingFileSink.<String>forRowFormat(new Path("./output1"), new SimpleStringEncoder<>())
                .withRollingPolicy(DefaultRollingPolicy.builder()
                        //按照文件大小滚动
                        //.withMaxPartSize(1*1024*1024*1024)
                        .withMaxPartSize(2 * 1024)
                        //按照时间滚动 每隔多久
                        .withRolloverInterval(15 * 60 * 1000)
                        //数据多久每来 滚动
                        .withInactivityInterval(5 * 60 * 1000)
                        .build())
                .build();

        streamSource.addSink(sink);
        env.execute();
    }
}
